package top.lingkang.potato.utils.rx;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.schedulers.Schedulers;
import javafx.application.Platform;

import java.util.concurrent.Executor;

/**
 * @author lingkang
 * @date 2021/9/19 23:06
 * @description
 */
public class RxObservableUtils<T> {
    private Observable<T> observable;

    public RxObservableUtils(ObservableOnSubscribe<T> subscribe, Watch<T> watch) {
        Observable.create(subscribe)
                .subscribeOn(Schedulers.io()) //在子线程中执行
                .observeOn(Schedulers.from(new Executor() {
                    @Override
                    public void execute(Runnable command) {
                        // 将线程执行发往主线程更新UI
                        Platform.runLater(command);
                    }
                }))//在主线程接收执行更新UI
                .subscribe(watch);
    }

    public RxObservableUtils(ObservableOnSubscribe<T> subscribe, Observer<T> observer) {
        Observable.create(subscribe)
                .subscribeOn(Schedulers.io()) //在子线程中执行
                .observeOn(Schedulers.from(new Executor() {
                    @Override
                    public void execute(Runnable command) {
                        // 将线程执行发往主线程更新UI
                        Platform.runLater(command);
                    }
                }))//在主线程接收执行更新UI
                .subscribe(observer);
    }

    public RxObservableUtils(ObservableOnSubscribe<T> subscribe) {
        observable = Observable.create(subscribe)
                .subscribeOn(Schedulers.io()) //在子线程中执行
                .observeOn(Schedulers.from(new Executor() {
                    @Override
                    public void execute(Runnable command) {
                        // 将线程执行发往主线程更新UI
                        Platform.runLater(command);
                    }
                }));//在主线程接收执行更新UI
    }

    public RxObservableUtils(ByWatch<T> byWatch) {
        observable = Observable.create(byWatch)
                .subscribeOn(Schedulers.io()) //在子线程中执行
                .observeOn(Schedulers.from(new Executor() {
                    @Override
                    public void execute(Runnable command) {
                        // 将线程执行发往主线程更新UI
                        Platform.runLater(command);
                    }
                }));//在主线程接收执行更新UI
    }

    public RxObservableUtils watch(Watch<T> w) {
        observable.subscribe(w);
        return this;
    }
}